package com.arch.bigdata.hbase1x.atguigu.guliweibo;

import com.google.common.collect.Lists;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.NamespaceDescriptor;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Delete;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.HBaseAdmin;
import org.apache.hadoop.hbase.client.HConnection;
import org.apache.hadoop.hbase.client.HConnectionManager;
import org.apache.hadoop.hbase.client.HTableInterface;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.filter.CompareFilter;
import org.apache.hadoop.hbase.filter.RowFilter;
import org.apache.hadoop.hbase.filter.SubstringComparator;
import org.apache.hadoop.hbase.io.compress.Compression;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;


/**
 * @author pizhihui
 * @date 2020-08-06
 */
public class WeiboOp {


    private Configuration conf = HBaseConfiguration.create();

    /**
     * 表名的定义
     */
    // 微博内容的表名
    public static final byte[] TABLE_CONTENT = Bytes.toBytes("weibo:content");
    // 用户关系表的表名
    public static final byte[] TABLE_RELATIONS = Bytes.toBytes("weibo:realtions");
    // 微博收件箱的表名
    public static final byte[] TABLE_RECEIVE_CONTENT_EMAIL = Bytes.toBytes("weibo:receive_content_email");


    /**
     * namespace 命名空间的定义
     */
    public void initNamespace() {
        HBaseAdmin admin = null;
        try {
            admin = new HBaseAdmin(conf);
            NamespaceDescriptor descriptor = NamespaceDescriptor.create("weibo")
                    .addConfiguration("creator", "jinji")
                    .addConfiguration("create_time", System.currentTimeMillis() + "")
                    .build();
            admin.createNamespace(descriptor);

        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            if (null != admin) {
                try {
                    admin.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }
    }

    /**
     * 创建微博内容表
     * Table Name:weibo:content
     * RowKey:用户 ID_时间戳
     * ColumnFamily:info
     * ColumnLabel:标题
     * 内容
     * 图片 URL
     * Version:1 个版本
     */
    public void createTableContent() {

        HBaseAdmin admin = null;
        try {
            admin = new HBaseAdmin(conf);

            HTableDescriptor tableDescriptor = new HTableDescriptor(TableName.valueOf(TABLE_CONTENT));
            HColumnDescriptor columnDescriptor = new HColumnDescriptor(Bytes.toBytes("info"));

            columnDescriptor.setBlockCacheEnabled(true);
            columnDescriptor.setBlocksize(2097152);
            columnDescriptor.setCompressionType(Compression.Algorithm.SNAPPY);
            columnDescriptor.setMaxVersions(1);
            columnDescriptor.setMinVersions(1);

            tableDescriptor.addFamily(columnDescriptor);
            admin.createTable(tableDescriptor);

        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            if (null != admin) {
                try {
                    admin.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }

    }

    /**
     * 用户关系表
     * Table Name:weibo:relations0
     * RowKey:用户 ID
     * ColumnFamily:attends,fans
     * ColumnLabel:关注用户 ID，粉丝用户 ID
     * ColumnValue:用户 ID
     * Version：1 个版本
     */
    public void createTableRelations() {

        HBaseAdmin admin = null;
        try {
            admin = new HBaseAdmin(conf);

            HTableDescriptor tableDescriptor = new HTableDescriptor(TableName.valueOf(TABLE_RELATIONS));


            HColumnDescriptor attendsColumn = new HColumnDescriptor(Bytes.toBytes("attends"));
            attendsColumn.setBlockCacheEnabled(true);
            attendsColumn.setBlocksize(2097152);
            attendsColumn.setCompressionType(Compression.Algorithm.SNAPPY);
            attendsColumn.setMaxVersions(1);
            attendsColumn.setMinVersions(1);

            HColumnDescriptor fansColumn = new HColumnDescriptor(Bytes.toBytes("fans"));
            fansColumn.setBlockCacheEnabled(true);
            fansColumn.setBlocksize(2097152);
            fansColumn.setCompressionType(Compression.Algorithm.SNAPPY);
            fansColumn.setMaxVersions(1);
            fansColumn.setMinVersions(1);

            tableDescriptor.addFamily(attendsColumn);
            tableDescriptor.addFamily(fansColumn);

            admin.createTable(tableDescriptor);

        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            if (null != admin) {
                try {
                    admin.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }
    }

    /**
     * 创建微博收件箱表
     * Table Name: weibo:receive_content_email
     * RowKey:用户 ID
     * ColumnFamily:info
     * ColumnLabel:用户 ID-发布微博的人的用户 ID
     * ColumnValue:关注的人的微博的 RowKey
     * Version:1000
     */
    public void createTableReceiveContentEmail() {

        HBaseAdmin admin = null;
        try {
            admin = new HBaseAdmin(conf);

            HTableDescriptor tableDescriptor = new HTableDescriptor(TableName.valueOf(TABLE_RECEIVE_CONTENT_EMAIL));

            HColumnDescriptor infoColumn = new HColumnDescriptor(Bytes.toBytes("info"));
            infoColumn.setBlockCacheEnabled(true);
            infoColumn.setBlocksize(2097152);
            infoColumn.setCompressionType(Compression.Algorithm.SNAPPY);
            infoColumn.setMaxVersions(1);
            infoColumn.setMinVersions(1);

            tableDescriptor.addFamily(infoColumn);

            admin.createTable(tableDescriptor);

        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            if (null != admin) {
                try {
                    admin.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }
    }



    /**
     * 发布微博
     * a、微博内容表中数据+1
     * b、向微博收件箱表中加入微博的 Rowkey
     */
    public void publishContent(String uid, String content) {

        try (HConnection connection = HConnectionManager.createConnection(conf)) {

            // a、微博内容表中添加 1 条数据，首先获取微博内容表描述
            HTableInterface contentTbl = connection.getTable(TableName.valueOf(TABLE_CONTENT));

            long timestamp = System.currentTimeMillis();
            String rowKey = uid + "_" + timestamp;

            Put put = new Put(Bytes.toBytes(rowKey));
            put.addColumn(Bytes.toBytes("info"), Bytes.toBytes("content"), timestamp, Bytes.toBytes(content));

            contentTbl.put(put);


            //b、向微博收件箱表中加入发布的 Rowkey
            HTableInterface relationTbl = connection.getTable(TableName.valueOf(TABLE_RELATIONS));

            Get get = new Get(Bytes.toBytes(uid));
            get.addFamily(Bytes.toBytes("fans"));


            List<byte[]> fans = Lists.newArrayList();
            Result result = relationTbl.get(get);
            for (Cell cell : result.rawCells()) {
                fans.add(CellUtil.cloneQualifier(cell));
            }

            // 如果没有粉丝,就直接返回
            if (fans != null && fans.size() <= 0) {
                return;
            }

            // c、开始操作收件箱表
            HTableInterface recTbl = connection.getTable(TableName.valueOf(TABLE_RECEIVE_CONTENT_EMAIL));

            List<Put> putList = Lists.newArrayList();
            for (byte[] fan : fans) {
                Put fanPut = new Put(fan);
                fanPut.addColumn(Bytes.toBytes("info"), Bytes.toBytes("uid"), timestamp, Bytes.toBytes(rowKey));
                putList.add(fanPut);
            }
            recTbl.put(putList);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /**
     * 关注用户逻辑
     * a、在微博用户关系表中，对当前主动操作的用户添加新的关注的好友
     * b、在微博用户关系表中，对被关注的用户添加粉丝（当前操作的用户）
     * c、当前操作用户的微博收件箱添加所关注的用户发布的微博 rowkey
     */
    public void addAttends(String uid, String... attends) {
        if (attends == null || attends.length <= 0 || uid == null || uid.length() <= 0) {
            return;
        }

        try (HConnection connection = HConnectionManager.createConnection(conf)) {

            // 关系表处理
            HTableInterface relationTbl = connection.getTable(TableName.valueOf(TABLE_RELATIONS));

            List<Put> puts = new ArrayList<Put>();

            Put attendPut = new Put(Bytes.toBytes(uid));
            for (String attend : attends) {
                attendPut.add(Bytes.toBytes("attends"), Bytes.toBytes(attend), Bytes.toBytes(attend));

                Put fansPut = new Put(Bytes.toBytes(attend));
                fansPut.add(Bytes.toBytes("fans"), Bytes.toBytes(uid), Bytes.toBytes(uid));

                puts.add(fansPut);
            }
            puts.add(attendPut);

            relationTbl.put(puts);


            // 内容处理
            HTableInterface contentTbl = connection.getTable(TableName.valueOf(TABLE_CONTENT));
            Scan scan = new Scan();

            //用于存放取出来的关注的人所发布的微博的 rowkey
            List<byte[]> rowkeys = new ArrayList<byte[]>();

            for (String attend : attends) {
                RowFilter rowFilter = new RowFilter(CompareFilter.CompareOp.EQUAL, new SubstringComparator(attend + "_"));
                scan.setFilter(rowFilter);
                ResultScanner resultScanner = contentTbl.getScanner(scan);
                Iterator<Result> iterator = resultScanner.iterator();
                while (iterator.hasNext()) {
                    Result next = iterator.next();
                    for (Cell cell : next.rawCells()) {
                        rowkeys.add(CellUtil.cloneQualifier(cell));
                    }
                }
            }

            if (rowkeys.size() <= 0) {
                return;
            }

            //得到微博收件箱表的操作对象

            HTableInterface recTBL = connection.getTable(TableName.valueOf(TABLE_RECEIVE_CONTENT_EMAIL));
            List<Put> recPutList = Lists.newArrayList();
            for (byte[] rk : rowkeys) {
                Put put = new Put(Bytes.toBytes(uid));
                String rowkey = Bytes.toString(rk);
                String attendUID = rowkey.substring(0, rowkey.indexOf("_"));
                long timestamp = Long.parseLong(rowkey.substring(rowkey.indexOf("-") + 1));
                put.add(Bytes.toBytes("info"), Bytes.toBytes(attendUID), timestamp, rk);
                recPutList.add(put);
            }
            recTBL.put(recPutList);
        } catch (Exception e) {

        }
    }

    /**
     * 取消关注（remove)
     * a、在微博用户关系表中，对当前主动操作的用户删除对应取关的好友
     * b、在微博用户关系表中，对被取消关注的人删除粉丝（当前操作人）
     * c、从收件箱中，删除取关的人的微博的 rowkey
     */
    public void removeAttends(String uid, String... attends) {
        if (uid == null || uid.length() <= 0 || attends == null || attends.length <= 0) {
            return;
        }


        try (HConnection connection = HConnectionManager.createConnection(conf)) {


            //a、在微博用户关系表中，删除已关注的好友
            HTableInterface relationTbl = connection.getTable(TableName.valueOf(TABLE_RELATIONS));

            List<Delete> deleteList = Lists.newArrayList();

            Delete attendDelete = new Delete(Bytes.toBytes(uid));

            for (String attend : attends) {
                attendDelete.deleteColumn(Bytes.toBytes("attends"), Bytes.toBytes(attend));

                // b
                Delete fansDelete = new Delete(Bytes.toBytes(attend));
                fansDelete.deleteColumn(Bytes.toBytes("fans"), Bytes.toBytes(uid));
                deleteList.add(fansDelete);

            }

            deleteList.add(attendDelete);
            relationTbl.delete(deleteList);

            //c、删除取关的人的微博 rowkey 从 收件箱表中
            HTableInterface recTBL = connection.getTable(TableName.valueOf(TABLE_RECEIVE_CONTENT_EMAIL));
            Delete recDelete = new Delete(Bytes.toBytes(uid));

            for (String attend : attends) {
                recDelete.deleteColumn(Bytes.toBytes("info"), Bytes.toBytes(attend));
            }
            recTBL.delete(recDelete);

        } catch (Exception e) {

        }

    }

    /**
     * 获取微博实际内容
     * a、从微博收件箱中获取所有关注的人的发布的微博的 rowkey
     * b、根据得到的 rowkey 去微博内容表中得到数据
     * c、将得到的数据封装到 Message 对象中
     */
    public List<Message> getAttendsContent(String uid) {

        try(HConnection connection = HConnectionManager.createConnection(conf)) {

            HTableInterface recTBL = connection.getTable(TableName.valueOf(TABLE_RECEIVE_CONTENT_EMAIL));

            //a、从收件箱中取得微博 rowKey
            Get get = new Get(Bytes.toBytes(uid));
            get.setMaxVersions(5);

            List<byte[]> rowkeyList = Lists.newArrayList();
            Result result = recTBL.get(get);

            for (Cell cell : result.rawCells()) {
                rowkeyList.add(CellUtil.cloneQualifier(cell));
            }

            //b、根据取出的所有 rowkey 去微博内容表中检索数据
            HTableInterface contentTbl = connection.getTable(TableName.valueOf(TABLE_CONTENT));
            List<Get> getList = Lists.newArrayList();

            for (byte[] rk : rowkeyList) {
                getList.add(new Get(rk));
            }

            Result[] results = contentTbl.get(getList);

            List<Message> messageList = Lists.newArrayList();
            for (Result res : results) {

                for(Cell cell : res.rawCells()){
                    Message message = new Message();
                    String rowkey = Bytes.toString(CellUtil.cloneRow(cell));
                    String userid = rowkey.substring(0, rowkey.indexOf("_"));
                    String timestamp = rowkey.substring(rowkey.indexOf("_") + 1);
                    String content = Bytes.toString(CellUtil.cloneValue(cell));

                    message.setUid(userid);
                    message.setTimestamp(timestamp);
                    message.setContent(content);

                    messageList.add(message);
                }


            }

            return messageList;
        } catch (Exception e) {
            e.printStackTrace();
        }

        return null;
    }



}
